package org.kumas.bigdata.hdfs3.fsdirectory;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.Writable;
import org.kumas.bigdata.hdfs3.checkpoint.SnapShot;
import org.kumas.bigdata.hdfs3.constant.Constant;
import org.kumas.bigdata.hdfs3.rpc.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*************************************************
 * TODO_Kumas
 *  Author： KumasZhang
 *  DateTime： 2021-12-02 8:23
 *  Description： hdfs内存目录树简易实现
 **/
public class FSDirectory implements Writable {
    public static final File SNAP_SHOT = new File(SnapShot.class.getResource("/").getPath() + "fs_snapshot");

    private static final Logger LOG = LoggerFactory.getLogger(FSDirectory.class);

    private INodeDirectory rootDir;
    private INodeMap inodeMap;

    public FSDirectory() {
        inodeMap = INodeMap.getInstance();
        rootDir = inodeMap.get("/").asDirectory();
    }

    INode getNode(String path) {
        return inodeMap.get(path);
    }

    /*************************************************
     * TODO_Kumas
     *  Description： 列出目录下内容
     **/
    public Response ls(String path) {
        if (StringUtils.isBlank(path)) {
            return Response.fail("无效的路径参数");
        }
        INode iNode = inodeMap.get(path);

        if (iNode == null) {
            return Response.fail("路径不存在：" + path);
        }

        if (iNode.isFile()) {
            return Response.ok(path);
        }

        StringBuilder sb = new StringBuilder("ls ").append(path).append(Constant.LINE_BREAKER);
        if (iNode.isDirectory()) {
            for (INode node : iNode.asDirectory().getChildrenList()) {
                sb.append("\t");
                sb.append(node.name);
                sb.append(Constant.LINE_BREAKER);
            }
        }
        return Response.ok(sb.toString());
    }

    /*************************************************
     * TODO_Kumas
     *  Description： 下载文件给远程客户端
     **/
    public Response cat(String file) {
        INode iNode = inodeMap.get(file);
        if (iNode == null || iNode.isDirectory()) {
            return Response.fail("文件不存在");
        }
        Block[] blocks = iNode.asFile().getBlocks();
        ByteBuffer byteBuffer = ByteBuffer.allocate(blocks.length * Block.BLOCK_SIZE);
        for (Block block : blocks) {
            byteBuffer.put(block.getContent());
        }
        return Response.ok("", byteBuffer.array());
    }

    public Response mkdir(String path) {
        return mkdir(path, true);
    }

    public Response mkdir(String dir, boolean recursive) {
        String path = dir.replace("//", "/");
        int lastSlash = path.lastIndexOf('/');
        String name = path.substring(lastSlash + 1);
        if (lastSlash == -1 || !path.startsWith("/")) {
            return Response.fail("路径不合法：" + path);
        }
        INode curNode = inodeMap.get(path);
        if (curNode != null) {
            if (!curNode.isDirectory()) {
                return Response.fail("当前路径已存在同名文件:" + path);
            } else {
                return Response.fail("目录已经存在:" + path);
            }
        }
        if (!recursive) {
            String parentPath = path.substring(0, lastSlash);
            if (lastSlash == 0) {//根节点
                parentPath = "/";
            }
            INode iNode = inodeMap.get(parentPath);
            if (iNode == null) {
                return Response.fail("父节点不存在：" + path);
            }

            INodeDirectory child = new INodeDirectory(name, iNode);
            iNode.asDirectory().addChild(child);
            inodeMap.put(child);
        } else {
            String[] split = path.split("/");
            String parentPath = "/";
            for (int i = 1; i < split.length; i++) {
                INodeDirectory parent = (INodeDirectory) inodeMap.get(parentPath);
                String currentPath = (parent.getAbsolutePath() + "/" + split[i]).replace("//", "/");
                if (inodeMap.get(currentPath) == null) {
                    INode current = new INodeDirectory(split[i], parent);
                    inodeMap.put(current);
                    parent.addChild(current);
                }
                parentPath = currentPath;
            }
        }
        return Response.ok("创建目录成功：" + path);
    }

    public Response createFile(String dir, String fileName, byte[] content) {
        return createFile(dir, fileName, FsUtils.bytesToBlocks(content));
    }

    public Response createFile(String dir, String fileName, Block[] blocks) {
        // TODO_Kumas 注释：
        String checkPath = checkPath(dir, fileName);
        if (checkPath != null) return Response.fail(checkPath);

        INode parent = inodeMap.get(dir);
        INodeFile fileNode = new INodeFile(fileName, parent.asDirectory(), blocks);
        inodeMap.put(fileNode);
        parent.asDirectory().addChild(fileNode);
        return Response.ok("创建文件成功：" + dir + ":" + fileName);
    }

    private String checkPath(String dir, String fileName){
        INode parent = inodeMap.get(dir);
        if (parent == null) {
            return ("目录不存在：" + dir);
        }
        if (!parent.isDirectory()) {
            return ("指定路径不是文件夹：" + dir);
        }
        INode destNode = inodeMap.get(dir + fileName);
        if (destNode != null) {
            return ("目录下已存在同名节点:" + dir + fileName);
        }
        return null;
    }

    public String print(INodeDirectory directory) {
        StringBuilder sb = new StringBuilder();
        sb.append(directory.getAbsolutePath()).append(Constant.LINE_BREAKER);
        for (INode node : directory.getChildrenList()) {
            if (node.isDirectory()) {
                INodeDirectory dir = (INodeDirectory) node;
                sb.append(print(dir));
            } else {
                sb.append(node.getAbsolutePath()).append(Constant.LINE_BREAKER);
            }
        }
        return sb.toString();
    }

    public Response print() {
        return Response.ok(print(rootDir));
    }

    @Override
    public void write(DataOutput out) throws IOException {
        LOG.info("序列化内存目录树开始：================================================================");
        out.writeUTF("fs{");
        out.writeInt(inodeMap.size() - 1);//剔除根节点
        Iterator<String> iter = inodeMap.iterator();
        while(iter.hasNext()) {
            INode node = inodeMap.get(iter.next());
            LOG.info(node.toString());
            if (node.getParent() == null) {//跳过根节点
                continue;
            }
            out.writeBoolean(node.isFile());
            out.writeUTF(node.name);
            out.writeUTF(node.getParent().getAbsolutePath());//父节点绝对路径
            if (node.isFile()) {
                Block[] blocks = node.asFile().getBlocks();
                out.writeInt(blocks.length);
                for (Block block : blocks) {
                    block.write(out);
                }
            }
        }
        out.writeUTF("}fs");
        LOG.info("序列化内存目录树结束：=*******************************************************");
    }

    @Override
    public void readFields(DataInput in) throws IOException  {
        in.readUTF();//"fs{"
        int size = in.readInt();//inodeMap.size()
        for (int i = 0; i < size; i++) {
            boolean isFile = in.readBoolean();
            String name = in.readUTF();
            String parentPath = in.readUTF();
            LOG.info(parentPath);
            LOG.info(this.mkdir(parentPath, true).toString());
            INodeDirectory parent = inodeMap.get(parentPath).asDirectory();

            if (isFile) {
                int blockSize = in.readInt();//blocksize
                Block[] blocks = new Block[blockSize];
                for (int j = 0; j < blockSize; j++) {
                    Block block = new Block();
                    block.readFields(in);
                    blocks[j] = block;
                }
                LOG.info(createFile(parentPath, name, blocks).toString());
            } else {
                if (parent == null) return; //父节点为null，则当前节点为根节点
                LOG.info(mkdir(parentPath + "/" + name, false).toString());
            }
        }
        in.readUTF();//"}fs"
    }
}
